package com.amumu.drama.common.utils;

import cn.hutool.core.collection.CollectionUtil;
import com.amumu.drama.common.utils.http.StringTools;
import com.beust.jcommander.internal.Lists;
import com.google.common.collect.Maps;
import org.slf4j.MDC;

import java.lang.management.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 线程相关的工具类，业务代码请尽量都使用此处的线程池，提高线程利用率
 *
 * @author BlackCat
 * @since 2015-03-24
 */
public final class ThreadUtils {

    /** 用于获取系统线程相关状态信息 */
    public static final ThreadMXBean threadMXBean = ManagementFactory.getThreadMXBean();

    // 以下是线程池状态的输出格式

    private static final String executorStatFmt = "%-22s %-13s %-12s %-16s %-12s %-12s %-18s %-16s %-16s %-12s\n";

    private static final String executorStatHeader = String.format(executorStatFmt, "name", "activeCount", "poolSize", "maximumPoolSize", "queueSize", "taskCount", "completedTaskCount",
            "corePoolSize", "keepAliveTime", "coreTimeOut");

    // 以下是etna框架的基础线程池，在业务代码中建议使用，提高线程利用率


    /** 默认的业务线程池，使用嵌入式方式启动会复用jetty的线程池，容器内启动则会创建一个线程池。如果需要执行优先级较低、可部分丢弃的任务，建议使用background线程池 */
    private static Executor defaultThreadPool;


    /** 纳入监控的线程池列表 */
    private static List<Object> threadPoolsForMonitor = Lists.newArrayList(defaultThreadPool);


    private ThreadUtils() {
    }

    /**
     * 休眠指定的毫秒数，内部将异常处理了不会对外抛出
     *
     * @return 如果有抛出异常返回false
     */
    public static boolean sleep(long millis) {
        try {
            Thread.sleep(millis);
            return true;
        } catch (InterruptedException e1) {
            return false;
        }
    }

    /**
     * 休眠指定的时间，单位可自行执行，内部将异常处理了不会对外抛出
     *
     * @return 如果有抛出异常返回false
     */
    public static boolean sleep(long sleepTime, TimeUnit unit) {
        return sleep(unit.toMillis(sleepTime));
    }


    /**
     * 获取默认的业务线程池，额外启动的服务可共享使用，避免开过多的线程耗费CPU资源
     * 注意，如果想要后台运行的任务纳入异常监控机制，请使用{@link BackgroundTask}来执行任务
     */
    public static Executor getDefault() {
        return defaultThreadPool;
    }


    /**
     * 线程过滤，进行条件筛选
     */
    private static boolean _isFilted(ThreadInfo info, boolean nameFilter, String name, boolean onlyRunnable) {
        if (nameFilter) {
            if (!info.getThreadName()
                    .toLowerCase()
                    .contains(name.toLowerCase())) {
                return true;
            }
        }
        if (onlyRunnable) {
            return Thread.State.RUNNABLE != info.getThreadState() && Thread.State.BLOCKED != info.getThreadState();
        }
        return false;
    }

    /**
     * 打印线程信息，抄自threadInfo.toString()
     */
    public static String printThreadInfo(ThreadInfo info, int maxFrames) {
        StringBuilder sb = new StringBuilder("\"" + info.getThreadName() + "\"" + " Id=" + info.getThreadId() + " " + info.getThreadState());
        if (info.getLockName() != null) {
            sb.append(" on " + info.getLockName());
        }
        if (info.getLockOwnerName() != null) {
            sb.append(" owned by \"" + info.getLockOwnerName() + "\" Id=" + info.getLockOwnerId());
        }
        if (info.isSuspended()) {
            sb.append(" (suspended)");
        }
        if (info.isInNative()) {
            sb.append(" (in native)");
        }
        sb.append('\n');
        int i = 0;
        StackTraceElement[] stackTrace = info.getStackTrace();
        for (; i < stackTrace.length && i < maxFrames; i++) {
            StackTraceElement ste = stackTrace[i];
            sb.append("\tat " + ste.toString());
            sb.append('\n');
            if (i == 0 && info.getLockInfo() != null) {
                Thread.State ts = info.getThreadState();
                switch (ts) {
                    case BLOCKED:
                        sb.append("\t-  blocked on " + info.getLockInfo());
                        sb.append('\n');
                        break;
                    case WAITING:
                        sb.append("\t-  waiting on " + info.getLockInfo());
                        sb.append('\n');
                        break;
                    case TIMED_WAITING:
                        sb.append("\t-  waiting on " + info.getLockInfo());
                        sb.append('\n');
                        break;
                    default:
                        break;
                }
            }
            for (MonitorInfo mi : info.getLockedMonitors()) {
                if (mi.getLockedStackDepth() == i) {
                    sb.append("\t-  locked " + mi);
                    sb.append('\n');
                }
            }
        }
        if (i < stackTrace.length) {
            sb.append("\t...");
            sb.append('\n');
        }
        LockInfo[] locks = info.getLockedSynchronizers();
        if (locks.length > 0) {
            sb.append("\n\tNumber of locked synchronizers = " + locks.length);
            sb.append('\n');
            for (LockInfo li : locks) {
                sb.append("\t- " + li);
                sb.append('\n');
            }
        }
        sb.append('\n');
        return sb.toString();
    }

    /**
     * 一次提交多个任务，并行执行，当所有任务都执行完后再进行下一步，如果有异常则会抛出中断
     */
    public static void multi(MultiJob... pj) {
        // 由于latch.await();会处于线程休眠状态,即使cdl减到0,也不会及时唤醒,提高此线程的优先级有助于尽快往下走.
        Thread.currentThread()
                .setPriority(Thread.MAX_PRIORITY);
        if (pj.length > 0) {
            CountDownLatch latch = new CountDownLatch(pj.length);
            for (MultiJob j : pj) {
                j.latch = latch;
                getDefault().execute(j);
            }
            try {
                latch.await();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            for (MultiJob j : pj) {
                if (null != j.throwable) {
                    throw new RuntimeException(j.throwable);
                }
            }
        }
        Thread.currentThread()
                .setPriority(Thread.NORM_PRIORITY);
    }

    /**
     * 带名称的线程工厂类，提供创建线程的功能
     */
    public static class NamedThreadFactory implements ThreadFactory {

        /** 线程组 */
        protected final ThreadGroup group;

        /** 保证原子操作的整数 */
        protected final AtomicInteger threadNumber = new AtomicInteger(1);

        /** 名字前缀 */
        protected final String namePrefix;

        /** 默认优先级 */
        protected int priority = Thread.NORM_PRIORITY;

        /** 是否为守护线程 */
        protected boolean daemon = false;

        public NamedThreadFactory(String namePrefix, int priority, boolean daemon) {
            this(namePrefix);
            this.daemon = daemon;
            this.priority = priority;
        }

        public NamedThreadFactory(String namePrefix, int priority) {
            this(namePrefix);
            this.priority = priority;
        }

        public NamedThreadFactory(String namePrefix) {
            SecurityManager s = System.getSecurityManager();
            group = (s != null) ? s.getThreadGroup() : Thread.currentThread()
                    .getThreadGroup();
            this.namePrefix = namePrefix;
        }

        /**
         * 创建一个新的线程
         */
        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);
            t.setDaemon(daemon);
            t.setPriority(priority);
            return t;
        }

        public String getNamePrefix() {
            return namePrefix;
        }

        @Override
        public String toString() {
            return getNamePrefix();
        }
    }

    /**
     * 后台线程池任务，将MDC中的用户信息等也带上了，在日志中会记录是谁的操作触发的，方便回溯问题
     */
    public static abstract class BackgroundTask implements Runnable {

        private Map<String, String> mdc = MDC.getCopyOfContextMap(); // 由初始化任务的线程将这个信息写入

        @Override
        public final void run() {
            try {
                if (null != mdc) {
                    MDC.setContextMap(mdc);
                }
                process();
            } catch (Throwable ex) {
                // 执行过程中抛异常会发邮件报警
                String title = ex.getClass()
                        .getSimpleName() + ":BackgroundTask";
            } finally {
                if (null != mdc) {
                    MDC.clear(); // 清除当前线程中记录的TAG
                }
            }
        }

        protected abstract void process() throws Throwable;
    }

    /**
     * 后台任务（不保证执行顺序）执行器
     */
    public static class BackgroundTaskExecutor {

        private String name;

        private ThreadPoolExecutor executor;

        /** 当排队的任务数达到多少时，启动报警并不再接受新的任务，防止内存溢出，如果为0表示不启用限制，注意要开启了线程池监控组件{@link }才能起到限制作用 */
        private int queueLimit = 0;

        /** 当前线程池是否爆了，由{@link }扫描启动了保护 */
        private volatile boolean taskExceed = false;

        public BackgroundTaskExecutor(String name, int num) {
            this.name = name;
            this.executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(num, new NamedThreadFactory(name, Thread.MIN_PRIORITY));
        }

        public BackgroundTaskExecutor(String name, int num, int queueLimit) {
            this(name, num);
            this.queueLimit = queueLimit;
        }

        public void execute(BackgroundTask task) {
            // 当等待执行的后台任务数达到上限时，停止接收新任务，防止堆积任务过多爆内存，起保护作用
            if (taskExceed) {
                return;
            }
            executor.execute(task);
        }

        public String getName() {
            return name;
        }

        public int getQueueLimit() {
            return queueLimit;
        }

        public void setTaskExceed(boolean taskExceed) {
            this.taskExceed = taskExceed;
        }

        /**
         * 获取等待执行的后台任务数
         */
        public int getQueueSize() {
            return executor.getQueue()
                    .size();
        }
    }

    /**
     * 后台任务（可根据指定的关键对象，按提交顺序执行）执行器
     */
    public static class OrderedTaskExecutor {

        private String name;

        private ThreadPoolExecutor[] executors;

        /** 当排队的任务数达到多少时，启动报警并不再接受新的任务，防止内存溢出，如果为0表示不启用限制，注意要开启了线程池监控组件{@link }才能起到限制作用 */
        private int queueLimit = 0;

        /** 当前线程池是否爆了，由{@link }扫描启动了保护 */
        private volatile boolean taskExceed = false;

        public OrderedTaskExecutor(String name, int num) {
            this.name = name;
            this.executors = new ThreadPoolExecutor[num];
            for (int i = 0; i < num; i++) {
                this.executors[i] = (ThreadPoolExecutor) Executors.newFixedThreadPool(1, new NamedThreadFactory(name, Thread.NORM_PRIORITY));
            }
        }

        public OrderedTaskExecutor(String name, int num, int queueLimit) {
            this(name, num);
            this.queueLimit = queueLimit;
        }

        public void execute(Object key, BackgroundTask task) {
            // 当等待执行的后台任务数达到上限时，停止接收新任务，防止堆积任务过多爆内存，起保护作用
            if (taskExceed) {
                return;
            }
            // 根据key对象确定是由哪个固定的线程执行任务
            int k = Math.abs(Objects.hashCode(key)) % executors.length;
            executors[k].execute(task);
        }

        public String getName() {
            return name;
        }

        public int getQueueLimit() {
            return queueLimit;
        }

        public void setTaskExceed(boolean taskExceed) {
            this.taskExceed = taskExceed;
        }

        /**
         * 获取等待执行的后台任务数
         */
        public int getQueueSize() {
            int count = 0;
            for (ThreadPoolExecutor executor : executors) {
                count += executor.getQueue()
                        .size();
            }
            return count;
        }
    }

    public static abstract class MultiJob implements Runnable {

        private CountDownLatch latch;

        private Throwable throwable;

        public abstract void process() throws Throwable;

        @Override
        public final void run() {
            try {
                process();
            } catch (Throwable t) {
                throwable = t;
            } finally {
                if (null != latch) {
                    latch.countDown();
                }
            }
        }
    }
}
